package com.wyl.learn04;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.StopWatch;
import com.rabbitmq.client.*;
import com.wyl.config.RabbitMqConfig;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * topic交换机
 */
public class Producer04_1 {
    public static final String EXCHANGE_NAME_TOPIC = "topic_exchange";

    public static void main(String[] argv) throws Exception {
        //创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(RabbitMqConfig.IP);
        factory.setUsername(RabbitMqConfig.USERNAME);
        factory.setPassword(RabbitMqConfig.PASSWORD);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME_TOPIC, BuiltinExchangeType.TOPIC);
        // 开启发布确认
        channel.confirmSelect();


        ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
        /**
         * 确认收到消息的一个回调
         * 1.消息序列号
         * 2.是否批量确认。true 可以确认小于等于当前序列号的消息；false 确认当前序列号消息。在RabbitMQ中，Confirm模式的批量确认是由服务器自动进行的，无法通过配置或参数设置来控制。
         */
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            if (multiple) {
                // 返回的是小于等于当前序列号的未确认消息 是一个 map
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag, true);
                System.out.println("成功接收消息：");
                System.out.println(CollUtil.join(confirmed.values(), "\n"));
                // 清除该部分未确认消息
                confirmed.clear();

            } else {
                // 只清除当前序列号的消息
                String message = outstandingConfirms.remove(deliveryTag);
                System.out.println("成功接收消息：" + message);
            }
        };
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            String message = outstandingConfirms.get(deliveryTag);
            System.out.println("发布的消息" + message + "未被确认，序列号" + deliveryTag);
        };
        /**
         * 添加一个异步确认的监听器
         * 1.确认收到消息的回调
         * 2.未收到消息的回调
         */
        channel.addConfirmListener(ackCallback, nackCallback);

        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        /**
         * Q1-->绑定的是
         *      中间带 orange 带 3 个单词的字符串(*.orange.*)
         * Q2-->绑定的是
         *      最后一个单词是 rabbit 的 3 个单词(*.*.rabbit)
         *      第一个单词是 lazy 的多个单词(lazy.#)
         */
        List<MessageBean> messageBeans = new ArrayList<>();
        messageBeans.add(new MessageBean("被队列 Q1Q2 接收到").setRouterKey("quick.orange.rabbit"));
        messageBeans.add(new MessageBean("被队列 Q1Q2 接收到").setRouterKey("lazy.orange.elephant"));
        messageBeans.add(new MessageBean("被队列 Q1 接收到").setRouterKey("quick.orange.fox"));
        messageBeans.add(new MessageBean("被队列 Q2 接收到").setRouterKey("lazy.brown.fox"));
        messageBeans.add(new MessageBean("虽然满足两个绑定但只被队列 Q2 接收一次").setRouterKey("lazy.pink.rabbit"));
        messageBeans.add(new MessageBean("不匹配任何绑定不会被任何队列接收到会被丢弃").setRouterKey("quick.brown.fox"));
        messageBeans.add(new MessageBean("是四个单词不匹配任何绑定会被丢弃").setRouterKey("quick.orange.male.rabbit"));
        messageBeans.add(new MessageBean("是四个单词但匹配 Q2").setRouterKey("lazy.orange.male.rabbit"));
        for (MessageBean messageBean : messageBeans) {
            String message = messageBean.getMessage();
            outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
            channel.basicPublish(EXCHANGE_NAME_TOPIC, messageBean.getRouterKey(), null, message.getBytes("UTF-8"));
        }
    }
}